package com.datasecops.scannerservices.operation;

import com.datasecops.scannerservices.db.DBHelper;
import com.datasecops.scannerservices.entity.CommonParam;
import com.datasecops.scannerservices.enumerate.DBType;
import com.datasecops.scannerservices.service.ScanDataProcessServer;
import com.datasecops.scannerservices.util.DBFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;

/**
 * HBase数据库操作类
 *
 * @param
 * @author SunLingzhi
 * @Description
 * @time 2022/10/21 18:13
 * @return
 */
@Service
@Slf4j
public class HBaseOperation {

    //每一批次定1000行
    private int batchLine = 1000;

    @Resource
    private ScanDataProcessServer dataProcessServer;


    /**
     * 获取数据库下面所有的表
     *
     * @param
     * @return
     * @Description
     * @author SunLingzhi
     * @time 2022/8/18 10:52
     */
    public List<String> getTablesByDBName(CommonParam param, String Instance) {
        String sourceKey = DBHelper.getSourceMapKey(param.getIp(), param.getPort(), param.getInstance());
        DBHelper dbHelper = DBFactory.getConnectDataSource(param.getTypeEnum(), param.getIp(), param.getPort(), param.getUserName(), param.getPassword(), Instance);
        if (null == dbHelper) {
            return null;
        }

        Connection connection = dbHelper.getConnection(sourceKey);
        Admin admin = null;
        List<String> tableNames = new ArrayList<>();
        try {
            admin = connection.getAdmin();
            TableName[] tables = admin.listTableNamesByNamespace(param.getDbName());
            Arrays.stream(tables).forEach(i -> tableNames.add(i.getNameAsString()));

        } catch (Exception e) {
            log.error("Failed Method HBaseOperation:: getTablesByDBName error :" + e.getMessage());
        } finally {
            if (null != admin) {
                try {
                    admin.close();
                } catch (IOException e) {
                    log.error("Failed Method HBaseOperation:: getTablesByDBName admin.close error :" + e.getMessage());
                }
            }
        }
        return tableNames;
    }

    /**
     * 获取DB数据返回
     *
     * @param
     * @param commonParam
     * @param typeEnum
     * @param jobId
     * @param jobName
     * @param scannerID
     * @param linage
     * @return
     * @Description
     * @author SunLingzhi
     * @time 2022/10/24 15:07
     */
    public boolean getDBResult(CommonParam param, CommonParam commonParam, DBType typeEnum, String jobId, String jobName, String scannerID, String linage) {
        String sourceKey = DBHelper.getSourceMapKey(param.getIp(), param.getPort(), param.getInstance());
        DBHelper dbHelper = DBFactory.getConnectDataSource(param.getTypeEnum(), param.getIp(), param.getPort(), param.getUserName(), param.getPassword(), param.getInstance());
        if (null == dbHelper) {
            return false;
        }
        Connection connection = dbHelper.getConnection(sourceKey);
        LinkedHashMap<String, LinkedList<String>> sourceDataMap = new LinkedHashMap<>();
        Table table = null;
        ResultScanner resultScanner = null;
        try {
            String tableName = param.getTableName();
            table = connection.getTable(TableName.valueOf(tableName));
            // 构建扫描器
            Scan scan = new Scan();
            //解决OOM问题(数据库量太大会内存溢出)
            //对行的优化
            scan.setCaching(batchLine);
            //对列的优化
            scan.setBatch(50);
            //开始行数
            //scan.setStartRow(Bytes.toBytes(0));
            //结束行数
            //scan.setStopRow(Bytes.toBytes(3));

            //limit
            if (null != param.getMaximumRows() && param.getMaximumRows() > 0) {
                scan.setLimit(param.getMaximumRows());
            }

            resultScanner = table.getScanner(scan);
            // 遍历数据并打印
            LinkedList<String> colsArray;
            String title;
            String val;
            int count = 0;
            int batch = 0;

            //处理tableName(HBase的表名为这种形式，需要剔除库名dsm:user)
            if (tableName.contains(":")){
                param.setTableName(tableName.split(":")[1]);
            }

            for (Result result : resultScanner) {
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    title = Bytes.toString(CellUtil.cloneQualifier(cell));
                    val = Bytes.toString(CellUtil.cloneValue(cell));

                    colsArray = sourceDataMap.getOrDefault(title, new LinkedList<>());
                    colsArray.add(val);

                    sourceDataMap.put(title, colsArray);
                }
                count++;
                if (count % batchLine == 0) {//每1000行，为一个批次
                    //数据操作
                    batch++;
                    dataProcessServer.tradingData(param, sourceDataMap, batch, commonParam, typeEnum, jobId, jobName,
                            scannerID, linage);
                    sourceDataMap.clear();
                }
            }
            if (sourceDataMap.size() > 0) {
                batch++;
                dataProcessServer.tradingData(param, sourceDataMap, batch, commonParam, typeEnum, jobId, jobName, scannerID, linage);
                sourceDataMap.clear();
            }
            resultScanner.close();
            table.close();
        } catch (Exception e) {
            log.error("Failed Method getDBResult error :" + e.getMessage());
            return false;
        } finally {
            if (null != resultScanner){
                resultScanner.close();
            }
            if (null != table){
                try {
                    table.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return true;
    }


    /**
     * 关闭连接
     * @Description
     * @author SunLingzhi
     * @time 2022/10/25 17:46
     * @param
     * @return
     */
    public void closeConnection(CommonParam param) {
        String sourceKey = DBHelper.getSourceMapKey(param.getIp(), param.getPort(), param.getInstance());
        DBHelper dbHelper = DBFactory.getConnectDataSource(param.getTypeEnum(), param.getIp(), param.getPort(), param.getUserName(), param.getPassword(), param.getInstance());
        if (null != dbHelper){
            Connection connection = dbHelper.getConnection(sourceKey);
            if (null != connection){
                try {
                    connection.close();
                } catch (IOException e) {
                    log.error("Failed Method closeConnection error :" + e.getMessage());
                }
            }
            dbHelper.removeSourceMap(sourceKey);
        }
    }

    public static void main(String[] args) {
        String ip = "10.26.18.202";
        int port = 0;
        String nameSpace = "dsm";

        CommonParam param1 = new CommonParam(DBType.HBASE, ip, port, null, null, nameSpace, null);
        HBaseOperation hBaseOperation = new HBaseOperation();

        List<String> tablesByDBName = hBaseOperation.getTablesByDBName(param1, null);

        CommonParam t2Param;
        for (String s : tablesByDBName) {
            t2Param = new CommonParam(null,null,DBType.HBASE, ip, port, null, null,nameSpace,s,null,null,null);
            //boolean dbResult = hBaseOperation.getDBResult(t2Param, commonParam);
           // System.out.println("end >> s:" + dbResult);
        }
    }
}